package com.atguigu.gmall.realtime.dwd.app;

/*import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import io.lettuce.core.ConnectionState;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

*//**
 * 交易域下单事务事实表
 *//*
public class DwdTradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(10014, 4, Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {

        //设置状态的TTL, 设置为5秒
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));



        // 读取topic_db的数据
        readOdsTopicDb(tableEnv, Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);


        // 过滤出四张表的数据
        // order_info               type = 'insert'
        String filterOrderInfo =
                " select " +
                        " `data`['id'] id," +
                        " `data`['user_id'] user_id," +
                        " `data`['province_id']  province_id" +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_info' " +
                        " and type = 'insert' ";
        // tableEnv.sqlQuery(filterOrderInfo).execute().print();
        Table orderInfoTable = tableEnv.sqlQuery(filterOrderInfo);
        tableEnv.createTemporaryView("order_info", orderInfoTable);

        // order_detail             type = 'insert'
        String filterOrderDetail =
                " select " +
                        " `data`['id'] id," +
                        " `data`['order_id'] order_id," +
                        " `data`['sku_id'] sku_id," +
                        " `data`['sku_name'] sku_name," +
                        " `data`['order_price'] order_price," +
                        " `data`['sku_num'] sku_num," +
                        " cast(cast(`data`['order_price'] as DECIMAL(16,2)) * cast(`data`['sku_num'] as DECIMAL(16,2)) as string) split_original_amount," +
                        " `data`['split_total_amount'] split_total_amount," +
                        " `data`['split_activity_amount'] split_activity_amount," +
                        " `data`['split_coupon_amount'] split_coupon_amount, " +
                        " `ts` " +
                        " from topic_db" +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_detail' " +
                        " and type = 'insert' ";
        // tableEnv.sqlQuery(filterOrderDetail).execute().print();
        Table orderDetailTable = tableEnv.sqlQuery(filterOrderDetail);
        tableEnv.createTemporaryView("order_detail", orderDetailTable);

        // order_detail_activity    type = 'insert'
        String filterActivity =
                " select " +
                        " `data`['order_detail_id'] order_detail_id," +
                        " `data`['activity_id'] activity_id," +
                        " `data`['activity_rule_id'] activity_rule_id " +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_detail_activity' " +
                        " and type = 'insert' ";
        // tableEnv.sqlQuery(filterActivity).execute().print();
        Table orderDetailActivityTable = tableEnv.sqlQuery(filterActivity);
        tableEnv.createTemporaryView("order_detail_activity", orderDetailActivityTable);

        // order_detail_coupon      type = 'insert'
        String filterCoupon =
                " select " +
                        " `data`['order_detail_id'] order_detail_id," +
                        " `data`['coupon_id'] coupon_id " +
                        " from topic_db" +
                        " where `database` = 'gmall` " +
                        " and `table` = 'order_detail_coupon' " +
                        " and type = 'insert' ";
        // tableEnv.sqlQuery(filterCoupon).execute().print();
        Table orderDetailCouponTable = tableEnv.sqlQuery(filterCoupon);
        tableEnv.createTemporaryView("order_detail_coupon", orderDetailCouponTable);


        // 常规 join 内连接 和 外连接  一般适用于正常的关联操作, 业务层面数据没有明确的先后,只需要考虑正常的传输延迟.
        // 间隔 join 一般适用于业务层面的数据有确定的先后顺序.
        // 维表(lookup) join 一般适用于维度关联
        // 此处我们使用 常规join
        // 注意: 状态TTL的设置, 考虑两点: 1.正常的传输延迟 2.业务上的滞后关系
        // 目前我们不考虑业务上的滞后关系,只考虑正常的传输延迟
        // order_detail
        // inner join
        // order_info
        // left join
        // order_detail_activity
        // left join
        // order_detail_coupon

        String joinSql =
                " select " +
                        " od.id," +
                        " od.order_id," +
                        " od.sku_id," +
                        " od.sku_name," +
                        " od.order_price," +
                        " od.sku_num," +
                        " od.split_original_amount," +
                        " od.split_total_amount," +
                        " od.split_activity_amount," +
                        " od.split_coupon_amount," +
                        " oi.user_id," +
                        " oi.province_id," +
                        " oda.activity_id," +
                        " oda.activity_rule_id," +
                        " odc.coupon_id," +
                        " od.ts " +
                        " from order_detail od " +
                        " join order_info oi" +
                        " on od.order_id = oi.id" +
                        " left join order_detail_activity oda" +
                        " on od.id = oda.order_detail_id" +
                        " left join order_detail_coupon odc" +
                        " on od.id = odc.order_detail_id ";
        // tableEnv.sqlQuery(joinSql).execute().print();
        Table joinTable = tableEnv.sqlQuery(joinSql);

        // 将数据写出到Kafka
        String createSinkTable =
                " create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + " ( " +
                        " id STRING," +
                        " order_id STRING," +
                        " sku_id STRING," +
                        " sku_name STRING," +
                        " order_price STRING," +
                        " sku_num STRING,\n" +
                        " split_original_amount STRING," +
                        " split_total_amount STRING," +
                        " split_activity_amount STRING," +
                        " split_coupon_amount STRING," +
                        " user_id STRING," +
                        " province_id STRING," +
                        " activity_id STRING," +
                        " activity_rule_id STRING," +
                        " coupon_id STRING " +
                        " ts BIGINT " +
                        " PRIMARY KEY(id) NOT ENFORCED" +
                        " ) " + FlinkSqlUtil.getUpsertKafkaSinkDDl(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);

        tableEnv.executeSql(createSinkTable);

        joinTable.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);


    }
}*/

import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author WEIYUNHUI
 * @date 2024/8/5 9:43
 *
 * 交易域下单事务事实表
 */
public class

DwdTradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(10014 , 4 , Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }
    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {

        //设置状态的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        // 读取topic_db的数据
        readOdsTopicDb(tableEnv , Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);

        // 过滤出四张表的数据
        // order_info              type = 'insert'
        filterOrderInfo(tableEnv);


        // order_detail            type = 'insert'
        filterOrderDetail(tableEnv);

        // order_detail_activity   type = 'insert'
        filterOrderDetailActivity(tableEnv);

        // order_detail_coupon     type = 'insert '
        filterOrderDetailCoupon(tableEnv);

        // join 4 张表
        Table joinTable = join(tableEnv);

        //写出到Kafka
        writerToKafka(tableEnv, joinTable);

    }

    private static void writerToKafka(StreamTableEnvironment tableEnv, Table joinTable) {
        String createSinkTable =
                " create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + " (" +
                        " id STRING, " +
                        " order_id STRING, " +
                        " sku_id STRING, " +
                        " sku_name STRING, " +
                        " order_price STRING, " +
                        " sku_num STRING, " +
                        " split_original_amount STRING, " +
                        " split_total_amount STRING, " +
                        " split_activity_amount STRING, " +
                        " split_coupon_amount STRING, " +
                        " user_id STRING, " +
                        " province_id STRING, " +
                        " activity_id STRING, " +
                        " activity_rule_id STRING, " +
                        " coupon_id STRING, " +
                        " ts BIGINT , "  +
                        " PRIMARY KEY(id) NOT ENFORCED " +
                        ") " + FlinkSqlUtil.getUpsertKafkaSinkDDl(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL) ;

        tableEnv.executeSql(createSinkTable) ;

        joinTable.executeInsert( Constant.TOPIC_DWD_TRADE_ORDER_DETAIL ) ;
    }

    private static Table join(StreamTableEnvironment tableEnv) {
        //常规 join     一般适用于正常的关联操作， 业务层面数据没有明确的先后， 只需要考虑正常的传输延迟。
        //间隔 join     一般适用于业务层面的数据有确定的先后顺序。
        //维表 join     一般适用于维度关联

        //此处使用 常规join
        //注意: 状态的TTL的设置  ， 考虑两点:  1. 正常的传输延迟   2. 业务上的滞后关系
        // order_detail
        // inner join
        // order_info
        // left join
        // order_detail_activity
        // left join
        // order_detail_coupon

        String joinSql =
                " select "+
                        " od.id , " +
                        " od.order_id , " +
                        " od.sku_id , " +
                        " od.sku_name , " +
                        " od.order_price , " +
                        " od.sku_num , " +
                        " od.split_original_amount , " +
                        " od.split_total_amount," +
                        " od.split_activity_amount," +
                        " od.split_coupon_amount," +
                        " oi.user_id , " +
                        " oi.province_id ," +
                        " oda.activity_id , " +
                        " oda.activity_rule_id , " +
                        " odc.coupon_id ," +
                        " od.ts " +
                        //"now() as join_time" 当前时间
                        " from order_detail od  " +
                        " join order_info  oi " +
                        " on od.order_id = oi.id " +
                        " left join order_detail_activity oda " +
                        " on od.id = oda.order_detail_id " +
                        " left join order_detail_coupon odc " +
                        " on od.id = odc.order_detail_id " ;

        //tableEnv.sqlQuery( joinSql ).execute().print();
        Table joinTable = tableEnv.sqlQuery(joinSql);
        return joinTable;
    }

    private static void filterOrderDetailCoupon(StreamTableEnvironment tableEnv) {
        String filterCoupon =
                " select " +
                        " `data`['order_detail_id'] order_detail_id, " +
                        " `data`['coupon_id'] coupon_id " +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_detail_coupon' " +
                        " and type = 'insert'" ;

        //tableEnv.sqlQuery( filterCoupon).execute().print();
        Table orderDetailCouponTable = tableEnv.sqlQuery(filterCoupon);
        tableEnv.createTemporaryView("order_detail_coupon" , orderDetailCouponTable) ;
    }

    private static void filterOrderDetailActivity(StreamTableEnvironment tableEnv) {
        String filterActivity =
                " select " +
                        " `data`['order_detail_id'] order_detail_id,\n" +
                        " `data`['activity_id'] activity_id,\n" +
                        " `data`['activity_rule_id'] activity_rule_id " +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_detail_activity' " +
                        " and type = 'insert'" ;
        //tableEnv.sqlQuery( filterActivity).execute().print();
        Table orderDetailActivityTable = tableEnv.sqlQuery(filterActivity);
        tableEnv.createTemporaryView("order_detail_activity" , orderDetailActivityTable);
    }

    private static void filterOrderDetail(StreamTableEnvironment tableEnv) {
        String filterOrderDetail =
                " select "+
                        " `data`['id'] id,\n" +
                        " `data`['order_id'] order_id,\n" +
                        " `data`['sku_id'] sku_id,\n" +
                        " `data`['sku_name'] sku_name,\n" +
                        " `data`['order_price'] order_price,\n" +
                        " `data`['sku_num'] sku_num,\n" +
                        "  cast( cast(`data`['order_price'] as decimal(16,2) ) * cast(`data`['sku_num'] as decimal(16,2) ) as string)split_original_amount,\n" +
                        " `data`['split_total_amount'] split_total_amount,\n" +
                        " `data`['split_activity_amount'] split_activity_amount,\n" +
                        " `data`['split_coupon_amount'] split_coupon_amount , " +
                        " ts " +
                        "  from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_detail' " +
                        " and type = 'insert'" ;
        //tableEnv.sqlQuery(filterOrderDetail).execute().print();
        Table orderDetailTable = tableEnv.sqlQuery(filterOrderDetail);
        tableEnv.createTemporaryView("order_detail" , orderDetailTable);
    }

    private static void filterOrderInfo(StreamTableEnvironment tableEnv) {
        String filterOrderInfo =
                " select " +
                        " `data`['id'] id , " +
                        " `data`['user_id'] user_id , " +
                        " `data`['province_id'] province_id " +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'order_info' " +
                        " and type = 'insert'" ;
        //tableEnv.sqlQuery(filterOrderInfo).execute().print();
        Table orderInfoTable = tableEnv.sqlQuery(filterOrderInfo);
        tableEnv.createTemporaryView("order_info" , orderInfoTable);
    }
}
